package com.xs.client.handler;

import java.nio.charset.Charset;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.xs.CommandType;
import com.xs.CustomProtcol;
import com.xs.client.entity.Home;
import com.xs.client.entity.RpcFuture;
import com.xs.client.entity.RpcRequest;
import com.xs.client.entity.RpcResponse;
import com.xs.client.entity.User;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

public class ClientHandler extends ChannelInboundHandlerAdapter {

	private static final Logger logger = Logger.getLogger(ClientHandler.class);

	private ChannelHandlerContext ctx;

	private Map<String, RpcFuture> responseMap = new ConcurrentHashMap<>();

	/**
	 * 
	 */
	public ClientHandler() {

	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {

		logger.info("客户端线程为channelActive:" + Thread.currentThread().getName());

		this.ctx = ctx;

		for (int i = 10; i < 1; i++) {

			CustomProtcol msg = new CustomProtcol();

			User u = new User();
			u.setId(i);
			u.setUserName(" 客户端的线程是 " + Thread.currentThread().getName() + "; user name is u" + i);
			Home h = new Home();
			h.setId(i);
			h.setAddress("my address is " + i);
			u.setHome(h);

			msg.setCommand(CommandType.READ);

			msg.setBody(JSON.toJSONString(u).getBytes(Charset.forName("UTF-8")));

			ctx.channel().writeAndFlush(msg);

			msg.setCommand(CommandType.CLOSE);

			ctx.channel().writeAndFlush(msg);

			/*
			 * ChannelPromise promise = ctx.newPromise();
			 * ctx.channel().writeAndFlush(msg,promise); promise.addListener(new
			 * GenericFutureListener<Future<? super Void>>() {
			 * 
			 * @Override public void operationComplete(Future<? super Void>
			 * future) throws Exception { if(future.isSuccess()) {
			 * 
			 * logger.info("client send msg success" );
			 * 
			 * 
			 * }
			 * 
			 * } });
			 */

		}

	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

		if (evt instanceof IdleStateEvent) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state().equals(IdleState.READER_IDLE)) {
				logger.info("READER_IDLE");
			} else if (event.state().equals(IdleState.WRITER_IDLE)) {
				/** 发送心跳,保持长连接 */
				String s = "ping$_";
				ctx.channel().writeAndFlush(s);
				logger.info("心跳发送成功!");
			} else if (event.state().equals(IdleState.ALL_IDLE)) {
				logger.info("ALL_IDLE");
			}
		}
		super.userEventTriggered(ctx, evt);

	}

	public RpcFuture sendMessage(CustomProtcol<RpcRequest> msg) throws InterruptedException, ExecutionException {

		if (ctx == null)
			throw new IllegalStateException();

		ChannelPromise promise = ctx.writeAndFlush(msg).channel().newPromise();

		RpcFuture future = new RpcFuture(msg.getBody(), promise);
		responseMap.put(msg.getBody().getRequestId(), future);

		logger.debug("ctx.writeAndFlush(msg).channel().newPromise() ：" + promise.isSuccess() + promise);

		/*
		 * promise.addListener(f -> { if (f.isSuccess()) { latch.countDown();
		 * logger.info("客户发消息已经发送完毕 sendMessage addListener promise ：" +
		 * f.isSuccess() + f);
		 * 
		 * } });
		 * 
		 * latch.await();
		 */
		logger.debug("客户发消息已经发送完毕 sendMessage promise ：" + promise.isSuccess() + promise);

		// logger.info("客户发消息已经发送完毕：" + msg.getBody());
		// logger.info("客户发送线程的promise:" + promise);

		return future;
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

		if (msg instanceof CustomProtcol) {
			@SuppressWarnings("unchecked")
			CustomProtcol<RpcResponse> cuc = (CustomProtcol<RpcResponse>) msg;
			logger.debug("客户端channelRead:" + cuc);
			RpcResponse response = cuc.getBody();
			// RpcResponse response = JSON.parseObject(data, RpcResponse.class);
			RpcFuture future = this.responseMap.get(response.getResponseId());
			if (future != null) {
				responseMap.remove(response.getResponseId());
				logger.debug("客户端future != null: before future.getPromise().isSuccess() = "
						+ future.getPromise().isSuccess());
				logger.debug(
						"客户端future != null: future.getPromise().setSuccess() before future.getPromise().isDone() = "
								+ future.getPromise().isDone());
				future.getPromise().setSuccess();
				future.setRpcResponse(response);
				logger.debug("客户端future != null: future.getPromise().setSuccess() after future.getPromise().isDone() = "
						+ future.getPromise().isDone());
				logger.debug("客户端future != null: after future.getPromise().isSuccess() = "
						+ future.getPromise().isSuccess());
				// logger.info("客户端收到信息的promise:" + future.getPromise());

			} else {
				return;
			}

			if (cuc.getCommand() == CommandType.CLOSE) {
				ctx.close();
			}

		}

	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		cause.printStackTrace();
		logger.error(cause);
		if (ctx != null) {
			ctx.close();
		}
	}

}
